package com.cet.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

public class FlowProcessorDemo {

    public static void main(String[] args) throws InterruptedException {

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // 中间操作
        TransformProcessor<String, String> transformProcessor1 = new TransformProcessor<>(item -> "hhh" + item);

        TransformProcessor<Object, String> transformProcessor2 = new TransformProcessor<>(item -> item + "======");


        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + ": 订阅开始了，" + subscription);
                this.subscription = subscription;

                // 请求一个数据
                subscription.request(1);
            }


            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + ": 订阅者接收到数据：" + item);
                if ("p-7".equals(item)) {
                    subscription.cancel();
                } else {
                    subscription.request(1);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + ": 订阅者接收到错误信号，" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + ": 管道处理完成");
            }
        };

        publisher.subscribe(transformProcessor1);
        transformProcessor1.subscribe(transformProcessor2);
        transformProcessor2.subscribe(subscriber);

        for (int i = 0; i < 10; i++) {
            publisher.submit("p-" + i);
        }

        publisher.close();

        TimeUnit.SECONDS.sleep(5);
    }
}

class TransformProcessor<T, R> extends SubmissionPublisher<R> implements Flow.Processor<T, R> {


    private Flow.Subscription subscription;

    private final Function<T, R> transformer;

    public TransformProcessor(Function<T, R> transformer) {
        this.transformer = transformer;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("processor 绑定完成");
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(T item) {
        System.out.println("processor 获取到数据：" + item);
        R transformedData = transformer.apply(item);
        // 将数据交给下游
        submit(transformedData);
        // 向上游请求数据
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}
